package com.hkbigdata.reallywar;

import com.hkbigdata.bean.MarketingUserBehavior;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @author liuanbo
 * @creat 2024-04-24-17:03
 * @see 2194550857@qq.com
 */
public class Flink04_Project_AppAnalysis_By_Chanel {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        DataStreamSource<MarketingUserBehavior> marketingUserBehaviorDataStreamSource = env.addSource(new AppMarketingDataSource());

        SingleOutputStreamOperator<Tuple2<String, Integer>> map = marketingUserBehaviorDataStreamSource.map(data -> Tuple2.of(data.getChannel() + "-" + data.getBehavior(), 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT));

        map.keyBy(data -> data.f0)
                .sum(1)
                .print();
        env.execute();


    }

    public static class AppMarketingDataSource extends RichSourceFunction<MarketingUserBehavior> {
        private boolean scan = true;
        private Random random = new Random();
        private List<String> behaviors = Arrays.asList("download", "install", "update", "uninstall");
        private List<String> channels = Arrays.asList("xiaomi", "apple", "huawei", "vivo");


        @Override
        public void run(SourceContext<MarketingUserBehavior> ctx) throws Exception {

            while (scan) {
                MarketingUserBehavior marketingUserBehavior = new MarketingUserBehavior(
                        (long) random.nextInt(10000),
                        behaviors.get(random.nextInt(behaviors.size())),
                        channels.get(random.nextInt(channels.size())),
                        System.currentTimeMillis()
                );
                ctx.collect(marketingUserBehavior);
                Thread.sleep(200);
            }
        }

        @Override
        public void cancel() {
            scan = false;
        }
    }
}
